package com.sinozo.data.app.function;


import com.sinozo.data.bean.TrafficNewUserBean;
import com.sinozo.data.common.DataConfig;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;


/**
 * @author zhaoyb
 * @version 1.0
 * @description 新用户水位线
 * @date 2023/5/18 15:40:45
 */
public class NewUserTimestampsAndWatermarks implements AssignerWithPeriodicWatermarks<TrafficNewUserBean> {

    //1小时,单位：毫秒
    private static final long MAX_OUT_OF_ORDERLINESS = DataConfig.MAX_OUT_OF_ORDERLINESS;

    private long currentMaxTimestamp;


    @Override
    public Watermark getCurrentWatermark() {

        return new Watermark(currentMaxTimestamp - MAX_OUT_OF_ORDERLINESS);
    }

    @Override
    public long extractTimestamp(TrafficNewUserBean element, long recordTimestamp) {
        long timestamp = element.getTs();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }
}
